package br.edu.ufcg.ccc.wordcount;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

@SuppressWarnings("deprecation")
public class ResultSorter {
	
	private static class ResultSorterMapper 
			extends MapReduceBase implements Mapper<LongWritable, Text, IntWritable, Text> {
		private Text text = new Text();
		
		public void map(LongWritable key, Text value,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			
			String[] tuple = splitPair(line);
			if (tuple != null) {
				IntWritable count = new IntWritable(Integer.parseInt(tuple[1]));
				text.set(tuple[0]);
				output.collect(count, text);
			}
		}
		
		private String[] splitPair(String line) {
			String[] tuple = new String[2];
			
			int separatorIndex = line.lastIndexOf('\t');
			if (separatorIndex < 0) return null;
			
			tuple[0] = line.substring(0, separatorIndex).trim();
			tuple[1] = line.substring(separatorIndex + 1).trim();
			
			return tuple;
		}
	}
	
	private static class ResultSorterReduce extends MapReduceBase implements Reducer<IntWritable, Text, IntWritable, Text> {
		public void reduce(IntWritable key, Iterator<Text> values,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
				throws IOException {
			
			while (values.hasNext()) {
				output.collect(key, values.next());
			}
		}
	}

	public static void main(String[] args) throws Exception {
		JobConf conf = new JobConf(ResultSorter.class);
		conf.setJobName("TWC Result SORTER");

		conf.setOutputKeyClass(IntWritable.class);
		conf.setOutputValueClass(Text.class);
		conf.setMapperClass(ResultSorterMapper.class);
		conf.setCombinerClass(ResultSorterReduce.class);
		conf.setReducerClass(ResultSorterReduce.class);
		
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		FileInputFormat.setInputPaths(conf, new Path(args[0]));
		FileOutputFormat.setOutputPath(conf, new Path(args[1]));

		JobClient.runJob(conf);
	}
}
